Ordering Requests: Part I

Requirements for order#

Replica coordination requires (alongside the agreement property) the order property, which states that every non-faulty state machine replica processes requests in the same relative order. One way to implement order is to assign unique identifiers to each request. This way, we can manage the order in which replicas process requests in the unique identifiers' total order.

With replicas processing requests according to the order of their unique identifiers, we define a request as stable at a state machine replica $sm_i$ if a non-faulty client cannot deliver a request with a lower unique identifier to $sm_i$. We can enforce the order requirement if every replica processes stable requests with the smallest unique identifier. This is called order implementation. The order implementation will require some request handling that enables replicas to maintain a list (or any data structure) of unprocessed stable requests and remove processed requests from that list.

Order implementation requires a method to assign unique identifiers to requests. This method also needs to be constrained by the assumptions that a client can make about the order in which state machines process requests:

1. [$O1$] A state machine processes requests by one of its clients in the order the client issued the requests. For example, a client $c$ issues requests to a state machine $sm$ in the following order: request $A$, then request $B$, and then request $C$. In this case, $sm$ will process $A$ first, then $B$, and then $C$.

2. [$O2$] If a client $c_0$​ makes a request $r_0$​ to state machine $sm$ that causes another client $c_1$​ to make a request $r_1$​ to $sm$, then $sm$ will process $r_0$​ before $r_1$​.

These two assumptions do not necessarily mean that $sm$ will process requests in the order they were made or received.

Let's discuss three ways in which we can implement order implementation. Replicas will need a test to classify a request as stable. We will have a stability test section for all our order implementations where we explain how replicas can check whether a request is stable.

Note: You can review how unique identifiers can be generated in the Sequencer chapter.

Order implementation#

Method 1: Logical clocks#

A logical clock is a function that maps events to integers. A logical clock $\hat{T}$ maps an event $e$ to a number $\hat{T}(e)$. It maps events such that for any two distinct events $e$ and $e'$, $\hat{T}(e)$ and $\hat{T}(e')$ cannot have the same value—either $\hat{T}(e)$ is greater than or lesser than $\hat{T}(e')$. Furthermore, if $e$ causes $e'$, $\hat{T}(e)$ is less than $\hat{T}(e')$.

Implementation#

We can implement logical clocks in a distributed system by associating a counter $\hat{T}_p$ with each process $p$. When $p$ sends a message, it includes a timestamp, which is the value of $\hat{T}_p$ when $p$ sends the message. The value of $\hat{T}_p$ is updated as follows:

1. Process $p$ increases $\hat{T}_p$ after each event at $p$.

2. When $p$ receives a message with a timestamp $\tau$, $p$ resets $\hat{T}_p$ based on the following equation:

The figure above shows how our implementation works with three processes $p$, $q$, and $r$. The dots represent events. Arrows connect dots such that the event at the tail might be responsible for causing the event at the head. For example, an arrow starts from an event from one processor and ends on an event from a different processor, the starting event is the sending of a message, and the ending event is the receiving of that message. The value of $\hat{T}_p(e)$ for each event is above its dot.

For an event $e$ that occurs at processor (node) $p$, let $\hat{T}(e)$ be a fixed length bitstring that uniquely identifies $p$ appended to the value of $\hat{T}_p(e)$ when $e$ occurs. $\hat{T}(e)$will uniquely identify an event $e$ such that $O1$ and $O2$ are satisfied.

Stability test#

Now that we understand the use of logical clocks for order implementation,  let's devise a stability test to allow replicas to test whether a request is stable.

Note: The method of using logical clocks for order implementation will not work in an asynchronous environment (where network messages can take arbitrarily long to be delivered and nodes can take arbitrarily long to process, and hence it is impossible to detect if a node has failed). Additionally, the methods of this subsection also don't apply to Byzantine failures. This section only applies to fail-stop failures where messages take a bounded time to deliver on the network.

We will require a mechanism to determine if a node or message has been delayed for a particular duration to classify it as faulty. We will discuss failure detection for fail-stop failures and make the following assumptions:

1. FIFO channels assumption: Messages between nodes are delivered in the order the sender sent them. This is equivalent to saying that nodes have FIFO channels between them.

2. Failure detection assumption: A node $p$ can only detect that another node $q$ has failed after receiving the last message sent by $q$ to $p$.

The second assumption is consistent with the first since fail-stop failure will occur at a node after it has sent its last message, which its recipient will receive after all other messages.

Based on the above assumptions, we can use the following stability test:

1. Every client makes a null request to the state machine after a set time.

2. A request will be stable at replica $sm_i$ if it has received a request with a larger timestamp from every non-faulty client.

Any client $c$ that $sm_i$ deems non-faulty will have a larger unique identifier than any of the previous requests made by $c$. Due to the FIFO channels assumption, $sm_i$ cannot receive a request with a smaller unique identifier from a non-faulty client. Therefore, all unique identifiers of requests that $sm_i$ receives after the current stable request will always be larger than the unique identifier of the stable request. The failure detection assumption suggests that $sm_i$ cannot receive a request from a failed client since its failure can only be detected after its last message has been sent, after which it will no longer send messages or requests. Therefore, $sm_i$ will not receive a request with a lower unique identifier than the unique identifier of a stable request.

Method 2: Synchronized real-time clocks#

Another way to generate unique identifiers is by using approximately synchronized real-time clocks. If $T_p(e)$ is the time (value on the real-time clock) at which event $e$ occurs at node $p$, we can create a unique identifier for $e$ by using this $T_p(e)$ and a unique fixed-length string to identify $p$—similar to how we did with logical clocks.

We'll implement the following restrictions to enforce the assumptions ($O1$ and $O2$) that a client can make about the order in which a state machine processes requests:

1. For $O1$, we assume a client cannot make two or more requests between successive clock ticks. For a resolution of $R$ seconds for processor clocks, every client can make at most one request every $R$ seconds. For two requests made by a client $c$ between successive clock ticks, a state machine could not tell which request was made first by $c$.

2. For $O2$, we assume clocks are synchronized to a better degree than the minimum message delivery time. If the difference in time on clocks on different nodes is within $\delta$ seconds, then it must take more than $\delta$ seconds for a message from a client to reach another. If we do not have this restriction, a client $c_1$ can request $r_1$ with a smaller unique identifier than a request $r_0$ that caused $c_1$ to make $r_1$.

Stability test#

We will formulate a stability test using the bounds on message delivery time. We will assume that $\Delta$ is a constant such that every non-faulty replica will receive a request $r$ no later than the time given by adding the request's unique identifier with $\Delta$ in the receiving replica's clock.

Once the clock on a node, say node $p$, reaches a time $\tau$, $p$ cannot receive a request such that the request's unique identifier is less than the difference between $\tau$ and $\Delta$ ($\tau-\Delta$).

A request $r$ will be stable at the state machine replica $sm_i$ if the clock of $sm_i$ reads $\tau$ such that the unique identifier of request $r$ is less than the difference of $\tau$ and $\Delta$ ($\tau-\Delta$).

The problem with this stability test is that it forces a state machine to lag behind its clients by $\Delta$, which is the worst-case message delivery time. We can avoid this by implementing the FIFO channel requirement mentioned earlier. With FIFO channels, a state machine replica can only receive requests with a unique identifier greater than previous requests. With this, we can have another stability test.

A request $r$ will be stable at $sm_i$ if a request with a larger unique identifier than $r$'s unique identifier has been received from every client.

The second test is not passed if a node refuses to make requests. We can combine the two tests, and a request is considered stable if it passes either test. In that case, the lag $\Delta$ only takes place if network delays or faulty nodes force it to occur.

Note: Placing an upper bound on the network messages is not an easy task. Google Spanner's TrueTime API achieved that feat by ensuring that clock times remain within a known bound.

What’s next?#

So far, we have discussed protocols for generating unique identifiers for requests where the client proposes the unique identifier. In the next lesson, we will discuss a protocol where state machine replicas propose and agree on a unique identifier for a request.